package cn.edu.nchu;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttListener implements Runnable {
    @Override
    public void run() {
        // 注册
        // 连接
        // 订阅
        String subTopic = "testtopic/#";
        int qos = 2;
        String broker = "ws://175.24.184.18:8083/mqtt";
        String clientId = "emqx_test";
        MemoryPersistence persistence = new MemoryPersistence();
        MqttClient client = null;
        try {
            client = new MqttClient(broker, clientId, persistence);

            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName("emqx_test");
            connOpts.setPassword("emqx_test_password".toCharArray());
            // 保留会话
            connOpts.setCleanSession(true);

            // 设置回调
            MqttClient finalClient = client;
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
                    try {
                        finalClient.reconnect();
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                                System.out.println("接收消息主题:" + topic);
             System.out.println("接收消息Qos:" + message.getQos());
            System.out.println("接收消息内容:" + new String(message.getPayload()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

                }
            });

            // 建立连接
            System.out.println("Connecting to broker: " + broker);
            client.connect(connOpts);

            // 订阅
            client.subscribe(subTopic);
//            client.subscribe(subTopic, (topic, payload) -> {
//                System.out.println("reply topic  : " + topic);
//                System.out.println("reply payload: " + payload.toString());
//            });



        } catch (MqttException me) {
            try {
                client.disconnect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
            System.out.println("Disconnected");
            try {
                client.close();
            } catch (MqttException e) {
                e.printStackTrace();
            }
            System.exit(0);
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        } finally {

        }
    }

    class MqttPostPropertyMessageListener implements IMqttMessageListener {
        @Override
        public void messageArrived(String var1, MqttMessage var2) throws Exception {
            System.out.println("reply topic  : " + var1);
            System.out.println("reply payload: " + var2.toString());
        }
    }
}
